package udf.matching;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Properties;

import org.apache.pig.EvalFunc;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.io.FileLocalizer;

import util.nlp.Dictionary;
import util.string.URLCleaner;

public class AnnotateURLSimple extends EvalFunc<Tuple> {

	boolean initialized = false;
	
	// static ArrayList<String> mTables = new ArrayList<String>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";

	public String init(Tuple tuple) throws IOException {
		/*
		 * for (int count = 1; count < tuple.size(); count++) { if
		 * (!(tuple.get(count) instanceof String)) { String msg =
		 * "LookupInFiles : Filename should be a string."; throw new
		 * IOException(msg); } mFiles.add((String) tuple.get(count)); }
		 */
		ArrayList<String> mFiles = new ArrayList<String>();
		
		String url = (String) tuple.get(0);
		
		
		String clean_url = URLCleaner.cleanURL(url);
		
		String domain_path = (String) tuple.get(1);
		String index_temp = (String) tuple.get(2);
		int index= Integer.valueOf(index_temp);
	
		mFiles.add(domain_path);
		
		Properties props = ConfigurationUtil
				.toProperties(PigMapReduce.sJobConfInternal.get());
		for (int i = 0; i < mFiles.size(); ++i) {
			// Files contain only 1 column with the key. No Schema. All keys
			// separated by new line.

			BufferedReader reader = null;

			InputStream is = null;
			try {
				is = FileLocalizer.openDFSFile(mFiles.get(i), props);
			} catch (IOException e) {
				String msg = "LookupInFiles : Cannot open file "
						+ mFiles.get(i);
				throw new IOException(msg, e);
			}
			try {
				reader = new BufferedReader(new InputStreamReader(is));
				String line;
				while ((line = reader.readLine()) != null) {
					String domain = line.split("\t")[index];// change this
					String domain_clean = URLCleaner.cleanURL(domain);
					
					// if we pass the candidate, url greater than  curren url in dict
					//return null
					if (clean_url.compareTo(domain)>0) 
						return null;
					
					if(domain_clean.startsWith(clean_url)){
						
						return domain; 
					}
					
					
				}

				is.close();
			} catch (IOException e) {
				String msg = "LookupInFiles : Cannot read file "
						+ mFiles.get(i);
				throw new IOException(msg, e);
			}
		}

		return null;
	}

	
	public Tuple exec(Tuple input) throws IOException {

		if (input.size() != 3 || input.get(0) == null) {
			return null;
		}

		else {
			try {
				Tuple output = TupleFactory.getInstance().newTuple(1);
				
				String url = (String) input.get(0);
				url = URLCleaner.cleanURL(url);
				
				String domain = init(input);
			//	System.out.println("matching: "  + url + "\t" + domain);
				output.set(0, domain);
				return output;

			} catch (Exception e) {
				System.err.println("Failed to process input; error - "
						+ e.getMessage());
				return null;
			}

		}

	}
	
	public static void main(String args[]){
		
		
		String a="ftp://adm.isp.at/ADM/ADMsmb-v0.2.tgz";
		String b="ftp://adm.isp.at/ADM/ADMsmb-v0.2.tgz";
		String c="c/adm.isp.at/ADM/ADMsmb-v0.2.tgz";
		
		
		System.out.println(a.compareTo(b));
		
	}
	

}